package kafka_day02

import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.log4j.{Level, Logger}
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010._
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

object KafkaDstreamdome {

  Logger.getLogger("org").setLevel(Level.ERROR)
  def main(args: Array[String]): Unit = {

    //创建streamingcontext
    val isLocal = args(0) toBoolean

    val conf: SparkConf = new SparkConf()
      .setAppName(this.getClass.getSimpleName)
    if (isLocal) {
      conf.setMaster("local[*]")
    }

    val ssc: StreamingContext = new StreamingContext(conf, Seconds(5))


    // kafka consumer相关的配置参数
    val kafkaParams = Map[String, String](
      "bootstrap.servers" -> "doit01:9092,doit02:9092,doit03:9092",
      // 反序列化的参数
      "key.deserializer" -> classOf[StringDeserializer].getName,
      "value.deserializer" -> classOf[StringDeserializer].getName,
      "group.id" -> "cp",
      "auto.offset.reset" -> "earliest",
      "enable.auto.commit" -> "false" //取消自动管理偏移量
    )

    val topics = Array(args(1))

    //创建Dstream(与kafka进行整合)
    //创建直连的数据流，使用新的API，效率更高
    val kafkaDStream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream(
      //Streamingcontext
      ssc,
      //如果Spark和Kafka在用一个阶段，任务优先调度到数据所在的机器上
      LocationStrategies.PreferConsistent,
      // 订阅主题
      ConsumerStrategies.Subscribe[String, String](topics, kafkaParams)
    )


    //需要获取kafka中的偏移信息
    //只有KafkaDStream中的KafkaRDD有偏移量信息，如果调用了Transformation算子，偏移量信息就不复存在了
    //只有第一手的RDD有偏移量信息，如果想获取Kafka的偏移量，只能通过KafkaUtils.createDirectStream返回的那个DStream获取
    //将kafkaDStream中的RDD依次遍历出来（目的是为了获取RDD中从kafka读取的偏移量）


    //遍历KafkaDStream中的RDD 信息
    kafkaDStream.foreachRDD(kafkaRDD => {

      //判断这个RDD是否为空
      if (!kafkaRDD.isEmpty()) {

        //从KafkaDStream中获取的RDD是kafkaRDD（第一手的RDD，有偏移量）
        //将RDD强转成HasOffsetRanges类型就可以获取到偏移量了

        val offsetRanges: Array[OffsetRange] = kafkaRDD.asInstanceOf[HasOffsetRanges].offsetRanges


        // 获取每一个分区的偏移量
        for (osr <- offsetRanges) {
          //topid
          val topic = osr.topic
          //分区编号
          val partition = osr.partition
          //起始偏移量
          val fromOffset = osr.fromOffset
          //结束偏移量
          val untilOffset = osr.untilOffset

          println(s"====> topic: $topic, partition: $partition, offset: $fromOffset -> $untilOffset")

        }

        kafkaRDD.map(_.value()).foreach(println)


        //将每一个分区的偏移量写入到Kafka中的那个特殊的topic（__consumer_offsets）
        //原来是5秒钟将偏移量更新到Kafka中，现在处理完一个批次的RDD更新一次偏移量
       // kafkaDStream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)



      }
println("*************************************************")
      println()



    })

    ssc.start()
    ssc.awaitTermination()



  }


}
